package es

import (
	"context"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"log"
	"os"
	"reflect"
	"strconv"
	"sync"

	"github.com/olivere/elastic/v7"

	"im/errcode"
)

var esClient *elastic.Client

var lock sync.Mutex

type Es interface {
	// 获取客户端
	GetClient() (*elastic.Client, error)
	// 创建或者更新索引
	TouchIndex(name string) error
	// 索引一个文档,可以传入struct,或者jsonString
	IndexDoc(indexName string, id int64, docJson interface{}) error
	// 索引一个文档,可以传入struct,或者jsonString
	IndexDocWait(indexName string, id int64, docJson interface{}, wait bool) error
	// 删除一个文档
	Delete(indexName string, id int64) error
	// 更新部分字段
	Update(indexName string, id int64, fields map[string]interface{}) error
	// 字段自增
	Increase(indexName string, id int64, field string) error
	// 字段自减
	Decrease(indexName string, id int64, field string) error
	UpdateByField(indexName string, script *elastic.Script, field string, value interface{}) error
}

func New() Es {
	return es{}
}

type es struct {
}

func Init() error {
	var err error
	client := New()
	err = client.TouchIndex("posts")
	if err != nil {
		return err
	}
	err = client.TouchIndex("replys")
	if err != nil {
		return err
	}
	err = client.TouchIndex("groups")
	return err
}
func (e es) findMappings(indexName string) string {
	base := "./dao/es/mappings"
	handle, err := os.Open(fmt.Sprintf("%s/%s_mappings.json", base, indexName))
	if err != nil {
		return ""
	}
	data, err := ioutil.ReadAll(handle)
	if err != nil {
		return ""
	}
	return string(data)
}
func (e es) Update(indexName string, id int64, docFields map[string]interface{}) error {
	client, err := e.GetClient()
	if err != nil {
		return err
	}
	_, err = client.Update().Index(WithPrefix(indexName)).
		Id(strconv.FormatInt(id, 10)).Doc(docFields).Do(context.Background())
	if err != nil {
		return errcode.NewError(errcode.EsError, err)
	}
	return nil
}
func (e es) UpdateByField(indexName string, script *elastic.Script, field string, value interface{}) error {
	client, err := e.GetClient()
	if err != nil {
		return err
	}
	_, err = client.UpdateByQuery(WithPrefix(indexName)).
		Query(elastic.NewTermsQuery(field, value)).
		Script(script).Do(context.Background())
	if err != nil {
		return errcode.NewError(errcode.EsError, err)
	}
	return nil
}

func (e es) Increase(indexName string, id int64, field string) error {
	client, err := e.GetClient()
	if err != nil {
		return err
	}
	_, err = client.Update().Index(WithPrefix(indexName)).
		Id(strconv.FormatInt(id, 10)).
		Script(elastic.NewScript(fmt.Sprintf("ctx._source.%s +=1", field))).
		Do(context.Background())
	if err != nil {
		return errcode.NewError(errcode.EsError, err)
	}
	return nil
}

func (e es) Decrease(indexName string, id int64, field string) error {
	client, err := e.GetClient()
	if err != nil {
		return err
	}
	_, err = client.Update().Index(WithPrefix(indexName)).
		Id(strconv.FormatInt(id, 10)).
		Script(elastic.NewScript(fmt.Sprintf("ctx._source.%s -=1", field))).
		Do(context.Background())
	if err != nil {
		return errcode.NewError(errcode.EsError, err)
	}
	return nil
}

func (e es) Delete(indexName string, id int64) error {
	client, err := e.GetClient()
	if err != nil {
		return err
	}
	_, err = client.Delete().Index(WithPrefix(indexName)).
		Id(strconv.FormatInt(id, 10)).Do(context.Background())
	if err != nil {
		return errcode.NewError(errcode.EsError, err)
	}
	return nil
}

func (e es) IndexDoc(indexName string, id int64, docJson interface{}) error {
	return e.IndexDocWait(indexName, id, docJson, false)
}
func (e es) IndexDocWait(indexName string, id int64, docJson interface{}, wait bool) error {
	client, err := e.GetClient()

	var doc string
	switch reflect.TypeOf(docJson).Kind() {
	case reflect.Struct:
		docBytes, _ := json.Marshal(docJson)
		doc = string(docBytes)
	case reflect.String:
		doc = docJson.(string)
	}
	svc := client.Index().Index(WithPrefix(indexName)).
		Id(strconv.FormatInt(id, 10)).BodyString(doc)
	if wait {
		svc = svc.Refresh("wait_for")
	}
	_, err = svc.Do(context.Background())
	if err != nil {
		return errcode.NewError(errcode.EsError, err)
	}
	return nil
}

func (e es) GetClient() (*elastic.Client, error) {
	var err error
	if esClient == nil {
		lock.Lock()
		options := []elastic.ClientOptionFunc{
			elastic.SetURL(fmt.Sprintf("http://%s:%s",
				os.Getenv("ES_HOST"), os.Getenv("ES_PORT"))),
			elastic.SetBasicAuth(os.Getenv("ES_USER"), os.Getenv("ES_PASS")),
		}
		if os.Getenv("APP_DEBUG") == "true" {
			options = append(options, elastic.SetTraceLog(log.New(os.Stdout, "es ", log.LstdFlags)), )
		}
		esClient, err = elastic.NewSimpleClient(options...)
		if err != nil {
			return nil, errcode.NewError(errcode.EsError, err)
		}
		lock.Unlock()
	}
	return esClient, nil
}
func (e es) TouchIndex(indexName string) error {
	client, err := e.GetClient()
	if err != nil {
		return err
	}
	ctx := context.Background()
	mapping := e.findMappings(indexName)
	exist, err := client.IndexExists(WithPrefix(indexName)).Do(ctx)
	if err != nil {
		return errcode.NewError(errcode.EsError, err)
	}
	// 存在则更新索引
	if exist {
		return e.updateMappings(WithPrefix(indexName), mapping)
	}

	// 新建索引
	_, err = client.CreateIndex(WithPrefix(indexName)).BodyString(mapping).Do(ctx)
	if err != nil {
		return errcode.NewError(errcode.EsError, err)
	}
	return nil
}
func (e es) updateMappings(indexNameWithPrefix string, mapping string) error {
	client, err := e.GetClient()
	if err != nil {
		return err
	}
	var mappingMap = map[string]interface{}{}
	err = json.Unmarshal([]byte(mapping), &mappingMap)
	if err != nil {
		return errcode.NewError(errcode.JsonParseError, err)
	}
	inner, _ := json.MarshalIndent(mappingMap["mappings"], "", "    ")
	_, err = client.PutMapping().Index(indexNameWithPrefix).BodyString(string(inner)).Do(context.Background())
	if err != nil {
		return errcode.NewError(errcode.EsError, err)
	}
	return nil
}
func WithPrefix(indexName string) string {
	return "im_" + indexName
}
